package com.baichen.trade.rocketmq;

import com.baichen.trade.exception.TradeMQException;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Date: 2019-11-07 23:33
 * @Author: baichen
 * @Description 消费者
 */
public class TradeMQConsumer {
    public static final Logger LOGGER = LoggerFactory.getLogger(TradeMQConsumer.class);
    private String topic;
    private String groupName;
    private String namesrvAddr;
    private String tag = "*";   // 标签，默认为* 多个tag以||分割
    private int consumeThreadMin = 20;
    private int consumeThreadMax = 64;

    private IMessageProcessor processor;

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getGroupName() {
        return groupName;
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public String getNamesrvAddr() {
        return namesrvAddr;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public IMessageProcessor getProcessor() {
        return processor;
    }

    public void setProcessor(IMessageProcessor processor) {
        this.processor = processor;
    }

    public void init() throws TradeMQException {
        if (StringUtils.isBlank(groupName)) {
            throw new TradeMQException("groupName is blank!");
        }
        if (StringUtils.isBlank(topic)) {
            throw new TradeMQException("topic is blank!");
        }
        if (StringUtils.isBlank(namesrvAddr)) {
            throw new TradeMQException("namesrvAddr is blank!");
        }
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.groupName);
        consumer.setNamesrvAddr(this.namesrvAddr);
        try {
            consumer.subscribe(this.topic, this.tag); // 主题和标签
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.setConsumeThreadMin(consumeThreadMin);
            consumer.setConsumeThreadMax(consumeThreadMax);
            TradeMessageListener listener = new TradeMessageListener();
            listener.setMessageProcessor(processor);
            consumer.registerMessageListener(listener);
            consumer.start();
            LOGGER.info(String.format("consumer is start! groupName:{%s},topic:{%s},namesrvAddr:{%s}", this.groupName, this.topic, this.namesrvAddr));
        } catch (MQClientException e) {
            LOGGER.error(String.format("consumer error! groupName:{%s},topic:{%s},namesrvAddr:{%s}", this.groupName, this.topic, this.namesrvAddr, e));
            throw new TradeMQException(e);
        }

    }
}
